// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_KV_SCAN_CONTEXT_HPP_
#define _XBASE_KV_SCAN_CONTEXT_HPP_

#include <boost/shared_ptr.hpp>
#include <boost/pool/detail/singleton.hpp>

#include <xthread/fast_lock.hpp>

#include <xbase_utility.hpp>
#include <protocol.hpp>
#include <logger.hpp>
#include <wrapper.hpp>
#include <kv.hpp>
#include <uuid.hpp>
#include <dynamic_buffer.hpp>
#include <column_name_parser.hpp>

#include "table_schema.hpp"

namespace xbase {

	// status
	class kv_scan_context {
	public:
		typedef boost::shared_ptr<kv_scan_context> pointer;

		struct tag {};
		typedef boost::details::pool::singleton_default< xbase::uuid<tag> > uuid;

	public:
		kv_scan_context (const std::string &p_tablet_name,
				 const std::string &p_start_row,
				 const std::string &p_end_row,
				 const std::vector<std::string> &p_columns,
				 uint64_t p_latest_timestamp,
				 uint64_t p_oldest_timestamp,
				 uint32_t p_ttl,
				 uint32_t p_num_version,
				 uint32_t p_options,
				 uint32_t p_num_row);

		// @p p_interval "interval time in seconds"
		bool is_expired (uint32_t p_interval) const {
			return (cur_seconds() - m_last_used_time) > p_interval;
		}

		// keep scanner alive
		void update_timestamp () {
			m_last_used_time = cur_seconds();
		}

		template<typename string_type>
		inline
		bool is_reached_end_row(const string_type &row) const {
			return ((! m_end_row.empty()) // not null end row and
				&& (row >= m_end_row));
		}
		
		template<typename string_or_column_name_parser_type>
		inline
		bool is_contain_column(const string_or_column_name_parser_type &column) const {
			if(is_contain_all_columns())
			{
				return true;
			}
			const std::vector<column_name_parser>::const_iterator end = m_column_name_parser.end();
			std::vector<column_name_parser>::const_iterator iter = m_column_name_parser.begin();
			for(; iter != end; ++iter)
			{
				if(iter->match(column))
				{
					return true;
				}
			}
			return false;
		}

		inline
		bool is_contain_timestamp (uint64_t timestamp) const {
			return ((! need_check_timestamp()) // 不需要检查时间戳
				||			   // 或者满足时间戳要求
				(((m_latest_timestamp == NO_TIMESTAMP) // 不检查
				  || (timestamp <= m_latest_timestamp))
				 &&
				 ((m_oldest_timestamp == NO_TIMESTAMP) // 不检查
				  || (timestamp >= m_oldest_timestamp))
				 &&
				 ((m_ttl == 0) // 不检查TTL
				  || (get_ttl(timestamp) <= m_ttl)))
				);
		}

		static inline
		uint64_t get_ttl(uint64_t timestamp) {
			const uint64_t cur = cur_micro_seconds();
			return (cur > timestamp) ? (cur - timestamp) : uint64_t(0);
		}

		// uint64_t timestamp() const {
		// 	return m_timestamp;
		// }

		bool need_check_timestamp() const {
			return m_need_check_timestamp;
		}

		// // 看是否传入的时间戳比指定时间戳更新
		// bool is_newer_timestamp(const uint64_t &p_timestamp) const {
		// 	return p_timestamp > m_timestamp;
		// }

		bool need_check_version() const {
			return m_need_check_version;
		}

		bool filter_deleted_enabled() const {
			return m_filter_deleted_enabled;
		}

		bool filter_deleted_itself_enabled() const {
			return m_filter_deleted_itself_enabled;
		}

		bool filter_value_disabled() const {
			return m_filter_value_disabled;
		}

		uint64_t id () const {
			return m_id;
		}

		std::string start_row() const {
			return m_start_row;
		}

		std::string end_row() const {
			return m_end_row;
		}

		uint32_t num_version() const {
			return m_num_version;
		}

		uint32_t num_row() const {
			return m_num_row;
		}

		std::string tablet_name() const {
			return m_tablet_name;
		}

		// std::string column() const {
		// 	return m_column;
		// }

		bool is_contain_all_columns() const {
			return m_is_contain_all_columns;
		}

		const std::vector<std::string> &groups() const {
			return m_groups;
		}

		bool has_bad_family() const {
			return (! m_bad_family.empty());
		}

		// bool is_any_column() const {
		// 	return m_column_name_parser.is_any_column();
		// }

		// std::string family_name() const {
		// 	return m_column_name_parser.family_name();
		// }

		// bool column_has_regex() const {
		// 	return m_column_name_parser.has_regex();
		// }

		// 根据扫描参数，构造一个最小的Key，使得通过这个Key找到的位置是
		// 离起始Key较近的前面几个位置或就是当前位置
		//
		// 构造好的数据放在给定的BUFFER中，返回key的指针
		//
		// 注意，给定的BUFFER要保证不再调整大小，否则会造成数据移动，使得返回的指针失效！
		const key_wrapper *get_least_key(simple_buffer &buffer) {
			std::string column;
			if(m_is_contain_all_columns)
			{
				column = "";
			}
			else
			{
				// 这时，最小的列中一定含有列族名
				if(m_least_column_name_parser.has_regex())
				{
					assert(m_least_column_name_parser.is_any_qualifier());
					// 在此保留列族名后面的冒号
					column = m_least_column_name_parser.family_name() + ':';
				}
				else
				{
					column = m_least_column_name_parser.column_name();
				}
			}
			
			uint64_t ts = ((m_latest_timestamp == NO_TIMESTAMP)
				       ? cur_micro_seconds()
				       : m_latest_timestamp);
			
			const std::string &start_row = m_start_row;
			boost::asio::mutable_buffer buf = buffer.prepare(start_row.size() + 4
									 + column.size() + 4
									 + sizeof(uint64_t)); // + 128);
											      // ensure it's large enough
			ByteBuffer bb(boost::asio::buffer_cast<char*>(buf),
				      boost::asio::buffer_size(buf),
				      boost::asio::buffer_size(buf));
			BinOArchive out(bb);
			out & start_row;
			out & column;
			out & ts;
		
			buffer.commit(bb.position());
		
			const key_wrapper *const least_key = reinterpret_cast<const key_wrapper*>(boost::asio::buffer_cast<const char*>(buf));
			return least_key;
		}

		std::string to_string() const;

	public:
		// option
		enum {
			OPTION_DISABLE_FILTER_DELETED = 0x00000001
			, OPTION_DISABLE_FILTER_DELETED_ITSELF = 0x00000002
			, OPTION_ENABLE_FILTER_VALUE = 0x00000004
		};

		// 第一位为“过滤删除记录”标记，0表示删除，1表示不删除
		static inline
		bool filter_deleted_option_enabled(uint32_t options) {
			return ((options & OPTION_DISABLE_FILTER_DELETED) == 0);
		}

		// 第二位为“是否过滤掉删除记录本身”标记，0表示过滤，1表
		// 示不过滤
		static inline
		bool filter_deleted_itself_option_enabled(uint32_t options) {
			return ((options & OPTION_DISABLE_FILTER_DELETED_ITSELF) == 0);
		}

		// 是否过滤掉值，即不返回记录的值
		static inline
		bool filter_value_option_disabled(uint32_t options) {
			return ((options & OPTION_ENABLE_FILTER_VALUE) == 0);
		}

	protected:
		table_schema::pointer get_schema() const;

	private:
		kv_scan_context(const kv_scan_context&);
		kv_scan_context &operator= (const kv_scan_context&);

	private:
		std::string m_table_name;
		mutable table_schema::weak_pointer m_schema;
		mutable FastLock m_schema_lock;

		// auto-generated
		uint64_t m_id;
		uint32_t m_last_used_time;
		mutable FastLock m_time_lock;

		// args
		
		std::string m_tablet_name;

		std::string m_start_row;
		std::string m_end_row;
		std::vector<std::string> m_columns;
		uint64_t m_latest_timestamp;
		uint64_t m_oldest_timestamp;
		uint64_t m_ttl;
		uint32_t m_num_version;
		uint32_t m_options; // 其它选项

		uint32_t m_num_row;

		const bool m_need_check_timestamp; // 是否需要检查时间戳
		const bool m_need_check_version;   // 是否需要检查版本数
		const bool m_filter_deleted_enabled; // 是否需要过滤掉
						     // 删除记录
		const bool m_filter_deleted_itself_enabled;
		// 是否过滤掉删除记录本身（只要最新的一条即可）；在合并
		// 紧缩的时候一定要设置该选项，才可以保留删除记录，使得
		// 后续写入的老版本的记录，也能被删除。
		const bool m_filter_value_disabled;

		std::vector<column_name_parser> m_column_name_parser; // 列名的解析结果
		bool m_is_contain_all_columns;
		column_name_parser m_least_column_name_parser; // 对应最小的列
		
		std::vector<std::string> m_groups; // 列名涉及到的所有组名
		std::vector<std::string> m_bad_family; // 找到的组名为空的列族名
	};
	
} // namespace xbase


#endif	// _XBASE_KV_SCAN_CONTEXT_HPP_
